package com.stars.easyms.mq.initializer;

import com.stars.easyms.base.batch.BatchResult;
import com.stars.easyms.mq.annotation.MQReceiver;
import com.stars.easyms.mq.annotation.MQSender;
import com.stars.easyms.mq.annotation.MQManager;
import com.stars.easyms.mq.asyn.AsynReceiveMQMessage;
import com.stars.easyms.mq.constant.MQConstants;
import com.stars.easyms.mq.core.*;
import com.stars.easyms.mq.dto.MQInfoDTO;
import com.stars.easyms.mq.enums.DestinationType;
import com.stars.easyms.mq.enums.RedeliveryPolicy;
import com.stars.easyms.mq.exception.EasyMsMQException;
import com.stars.easyms.mq.properties.EasyMsMQProperties;
import com.stars.easyms.mq.service.MQMessageService;
import com.stars.easyms.base.util.*;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ConfigurableApplicationContext;

import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.lang.reflect.Modifier;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * MQ服务初始化启动类
 *
 * @author guoguifang
 * @date 2018-04-23 13:54
 * @since 1.0.0
 */
public final class MQManagerInitializer {

    private static final Logger logger = LoggerFactory.getLogger(MQManagerInitializer.class);

    private static AtomicBoolean isInit = new AtomicBoolean(false);

    private ConfigurableApplicationContext applicationContext;

    private String applicationId;

    private EasyMsMQProperties easyMsMQProperties;

    private MQMessageService mqMessageService;

    /**
     * 有异常直接启动服务器失败
     */
    public void init() {
        if (!easyMsMQProperties.isEnabled()) {
            return;
        }
        // 判断是否异步启动
        if (easyMsMQProperties.isAsynLoad()) {
            ExecutorUtil.execute(this::initMQ);
        } else {
            initMQ();
        }
    }

    private void initMQ() {
        if (isInit.compareAndSet(false, true)) {
            logger.info("Begin to register MQ service......");
            try {
                initMQManager();
                logger.info("Register MQ service success!");
            } catch (Exception e) {
                logger.info("Register MQ service fail!", e);
                System.exit(1);
            }
        }
    }

    private void initMQManager() {
        // 获取所有加了MQSender、MQReceiver注解的方法
        Set<Method> mqManagerMethodSet = ReflectUtil.getAllMethodByAnnotationWithDefaultAll(easyMsMQProperties.getBasePackage(), MQSender.class, MQReceiver.class);
        if (mqManagerMethodSet == null || mqManagerMethodSet.isEmpty()) {
            if (logger.isDebugEnabled()) {
                logger.debug("Effective MQSender/MQReceiver method was not found！");
            }
            return;
        }

        for (Method mqManagerMethod : mqManagerMethodSet) {
            Class<?> mqManagerClass = mqManagerMethod.getDeclaringClass();
            String mqManagerClassName = mqManagerClass.getName();
            if (!mqManagerClass.isAnnotationPresent(MQManager.class)) {
                throw new EasyMsMQException("Class({}) without annotation(MQManager)!", mqManagerClassName);
            }

            // 获取方法的输入、输出参数类型
            Class<?>[] argumentTypes = mqManagerMethod.getParameterTypes();
            Class<?> returnType = mqManagerMethod.getReturnType();
            String mqManagerMethodName = mqManagerMethod.getName();
            // mq发送与接收的参数个数必须是一个
            if (argumentTypes.length != 1) {
                throw new EasyMsMQException("MQ service({}) method({}) parameter count must be one!", mqManagerClassName, mqManagerMethodName);
            }
            if (!Modifier.isPublic(mqManagerMethod.getModifiers())) {
                throw new EasyMsMQException("MQ service({}) method({}) modifier must be public!", mqManagerClassName, mqManagerMethodName);
            }

            Map<String, MQInfoDTO> mqInfoDTOMap = null;
            String type = null;
            String key = null;
            String name = null;
            DestinationType destinationType = null;
            int channelCount = 0;
            int channelSize = 0;
            long maxFailCount = 0L;
            long redeliveryDelay = 0L;
            Boolean batch = null;
            Boolean allowNonPersistent = null;
            Boolean overall = null;
            Boolean isDynamicKey = null;
            RedeliveryPolicy redeliveryPolicy = null;
            if (mqManagerMethod.isAnnotationPresent(MQSender.class)) {
                MQSender mqSender = mqManagerMethod.getAnnotation(MQSender.class);
                key = mqSender.key();
                name = mqSender.name();
                destinationType = mqSender.destinationType();
                channelCount = mqSender.channelCount();
                channelSize = mqSender.channelSize();
                maxFailCount = mqSender.maxFailCount();
                redeliveryPolicy = mqSender.redeliveryPolicy();
                redeliveryDelay = mqSender.redeliveryDelay();
                batch = mqSender.batch();
                isDynamicKey = mqSender.isDynamicKey();
                allowNonPersistent = mqSender.allowNonPersistent();
                mqInfoDTOMap = MQConstants.MQ_SENDER_INFO_MAP;
                type = "MQSender";
            } else if (mqManagerMethod.isAnnotationPresent(MQReceiver.class)) {
                Class<?>[] exceptionTypes = mqManagerMethod.getExceptionTypes();
                if (exceptionTypes.length != 1 || !exceptionTypes[0].isAssignableFrom(Exception.class)) {
                    throw new EasyMsMQException("MQ service({}) method({}) annotation(MQReceiver) must throws Exception!", mqManagerClassName, mqManagerMethodName);
                }
                MQReceiver mqReceiver = mqManagerMethod.getAnnotation(MQReceiver.class);
                key = mqReceiver.key();
                name = mqReceiver.name();
                destinationType = mqReceiver.destinationType();
                channelCount = mqReceiver.channelCount();
                maxFailCount = mqReceiver.maxFailCount();
                redeliveryDelay = mqReceiver.redeliveryDelay();
                redeliveryPolicy = mqReceiver.redeliveryPolicy();
                batch = mqReceiver.batch();
                overall = mqReceiver.overall();
                if (Boolean.TRUE.equals(batch) && !argumentTypes[0].isAssignableFrom(List.class)) {
                    throw new EasyMsMQException("MQ service({}) method({}) annotation(MQReceiver) when batch receive, the parameter type must be List!", mqManagerClassName, mqManagerMethodName);
                }
                if (Boolean.TRUE.equals(batch) && Boolean.FALSE.equals(overall)) {
                    if (!returnType.isAssignableFrom(BatchResult.class)) {
                        throw new EasyMsMQException("MQ service({}) method({}) annotation(MQReceiver) when batch receive and use non-overall, the return type must be BatchResult!", mqManagerClassName, mqManagerMethodName);
                    }
                } else {
                    if (!returnType.isAssignableFrom(Boolean.class) && !returnType.isAssignableFrom(boolean.class)) {
                        throw new EasyMsMQException("MQ service({}) method({}) annotation(MQReceiver) when batch receive and use overall or non-batch receive, the return type must be Boolean or boolean!", mqManagerClassName, mqManagerMethodName);
                    }
                }
                isDynamicKey = mqReceiver.isDynamicKey();
                mqInfoDTOMap = MQConstants.MQ_RECEIVER_INFO_MAP;
                type = "MQReceiver";
            }
            boolean isElExpression = key != null && (key.startsWith("${") && key.endsWith("}") || key.contains("."));
            if (isElExpression) {
                try {
                    key = PropertyPlaceholderUtil.replace(getReplaceKey(key));
                } catch (Exception e) {
                    throw new EasyMsMQException("MQ service({}) method({}) annotation({}) {}!", mqManagerClassName, mqManagerMethodName, type, e.getMessage());
                }
            }
            if (StringUtils.isBlank(key)) {
                throw new EasyMsMQException("MQ service({}) method({}) annotation({}) key is blank!", mqManagerClassName, mqManagerMethodName, type);
            }

            if (mqInfoDTOMap.containsKey(key)) {
                throw new EasyMsMQException("MQ service({}) method({}) annotation({}) key({}) already exist in the class({}) method({})!",
                        mqManagerClassName, mqManagerMethodName, type, key, mqInfoDTOMap.get(key).getClass().getName(), mqInfoDTOMap.get(key).getMethod().getName());
            }
            Class<?>[] parameterClasses = mqManagerMethod.getParameterTypes();
            if (parameterClasses.length != 1) {
                throw new EasyMsMQException("MQ service({}) method({}) parameter count must be one!", mqManagerClassName, mqManagerMethodName);
            }

            Object instance = applicationContext.getBean(mqManagerClass);
            Field[] fields = mqManagerClass.getDeclaredFields();
            for (Field field : fields) {
                if (!field.isAccessible()) {
                    field.setAccessible(true);
                }
                Object fieldValue = null;
                try {
                    fieldValue = field.get(instance);
                } catch (IllegalAccessException e) {
                    logger.warn("MQ service({}) field({}) value can't get!", mqManagerClassName, field.getName());
                }
                if (fieldValue == null && field.isAnnotationPresent(Autowired.class)) {
                    fieldValue = applicationContext.getBean(field.getType());
                    try {
                        field.set(instance, fieldValue);
                    } catch (IllegalAccessException e) {
                        logger.warn("MQ service({}) field({}) value can't set!", mqManagerClassName, field.getName());
                    }
                }
            }

            MQInfoDTO mqInfoDTO = new MQInfoDTO();
            if (Boolean.TRUE.equals(isDynamicKey)) {
                mqInfoDTO.setStaticKey(key);
                key = StringUtils.join(new Object[]{key, PropertyPlaceholderUtil.replace("spring.profiles.active", "default")}, "_");
            }
            mqInfoDTO.setKey(key);
            mqInfoDTO.setName(name);
            mqInfoDTO.setDestinationType(destinationType);
            mqInfoDTO.setChannelCount(channelCount <= 1 ? 1 : channelCount >= 5 ? 5 : channelCount);
            mqInfoDTO.setChannelSize(channelSize <= 1 ? 1 : channelSize >= 2000 ? 2000 : channelSize);
            mqInfoDTO.setMaxFailCount(maxFailCount <= 1 ? 1 : maxFailCount);
            mqInfoDTO.setRedeliveryPolicy(redeliveryPolicy);
            mqInfoDTO.setRedeliveryDelay(redeliveryDelay <= 0 ? 0 : redeliveryDelay);
            mqInfoDTO.setBatch(batch);
            mqInfoDTO.setOverall(overall);
            mqInfoDTO.setIsDynamicKey(isDynamicKey);
            mqInfoDTO.setAllowNonPersistent(allowNonPersistent);
            mqInfoDTO.setClazz(mqManagerClass);
            mqInfoDTO.setMethod(mqManagerMethod);
            mqInfoDTO.setParameterClass(parameterClasses[0]);
            mqInfoDTO.setParameterType(mqManagerMethod.getGenericParameterTypes()[0]);
            mqInfoDTO.setInstance(instance);
            mqInfoDTOMap.put(key, mqInfoDTO);
            logger.info("Found MQ service: {}:{} ==> {}.{}()", type, key, mqManagerClassName, mqManagerMethodName);
        }

        // 注册发送与接收多通道
        registMultichannel(MQConstants.MQ_SENDER_INFO_MAP, MQConstants.MQ_SENDER_CHANNEL_INFO_MAP);
        registMultichannel(MQConstants.MQ_RECEIVER_INFO_MAP, MQConstants.MQ_RECEIVER_CHANNEL_INFO_MAP);

        // 启动MQ调度器
        MQScheduler.setMQMessageService(mqMessageService);
        MQScheduler.setApplicationId(applicationId);
        MQScheduler.setMqMessageSenderFactory(new MQMessageSenderFactory(applicationContext, applicationId, easyMsMQProperties));
        int mqSendChannelInfoMapSize = MQConstants.MQ_SENDER_CHANNEL_INFO_MAP.size();
        int mqReceiveChannelInfoMapSize = MQConstants.MQ_RECEIVER_CHANNEL_INFO_MAP.size();
        int scheduleThreadCount = mqReceiveChannelInfoMapSize > 0 ? 3 : 2;
        ThreadFactory schedulerThreadFactory = new DefaultThreadFactory().setNameFormat("easyms-mq-scheduler-pool-%d").build();
        ExecutorService scheduleThreadPool = new ThreadPoolExecutor(scheduleThreadCount, scheduleThreadCount, 0, TimeUnit.SECONDS, new LinkedBlockingQueue<>(), schedulerThreadFactory);
        scheduleThreadPool.execute(MQScheduler.newMQSendSchedule(mqSendChannelInfoMapSize + 2, (mqSendChannelInfoMapSize + 1) * 3 + 2));
        if (mqReceiveChannelInfoMapSize > 0) {
            scheduleThreadPool.execute(MQScheduler.newMqReceiveScheduler(mqReceiveChannelInfoMapSize + 2, (mqReceiveChannelInfoMapSize + 1) * 3 + 2));

            // 启动MQ监听器
            ExecutorUtil.execute(() -> {
                AsynReceiveMQMessage asynReceiveMQMessage = new AsynReceiveMQMessage();
                for (String multichannelKey : MQConstants.MQ_RECEIVER_CHANNEL_INFO_MAP.keySet()) {
                    MQInfoDTO mqInfoDTO = MQConstants.MQ_RECEIVER_CHANNEL_INFO_MAP.get(multichannelKey);
                    List<MQListener> mqListenerList = mqInfoDTO.getMqListenerList();
                    if (mqListenerList == null) {
                        mqListenerList = new ArrayList<>();
                        mqInfoDTO.setMqListenerList(mqListenerList);
                    }
                    MQListenerFactory mqListenerFactory = new MQListenerFactory(applicationContext, applicationId, easyMsMQProperties, mqMessageService, asynReceiveMQMessage);
                    MQListener mqListener = mqListenerFactory.getMQListener(mqInfoDTO, multichannelKey);
                    mqListener.start();
                    mqListenerList.add(mqListener);
                }
            });
        }

        // 启动MQ清理器
        scheduleThreadPool.execute(MQMigrator.createInstance(mqMessageService));
    }

    private void registMultichannel(Map<String, MQInfoDTO> mqInfoMap, Map<String, MQInfoDTO> mqChannelInfoMap) {
        mqInfoMap.forEach((key, mqInfoDTO) -> {
            Integer channelCount = mqInfoDTO.getChannelCount();
            if (channelCount != null && channelCount > 1) {
                for (int i = 1; i <= channelCount; i++) {
                    String multichannelKey = StringUtils.join(new Object[]{key, StringUtils.leftPad(String.valueOf(i), 3, "0")}, "-");
                    mqChannelInfoMap.put(multichannelKey, mqInfoDTO);
                }
            } else {
                mqChannelInfoMap.put(key, mqInfoDTO);
            }
        });
    }

    private String getReplaceKey(String key) {
        if (key.startsWith("${") && key.endsWith("}")) {
            key = key.substring(2, key.length() - 1);
        }
        return EasyMsMQProperties.EASYMS_MQ_PREFIX + ".key." + key;
    }

    public MQManagerInitializer(ConfigurableApplicationContext applicationContext, String applicationId, EasyMsMQProperties easyMsMQProperties, MQMessageService mqMessageService) {
        this.applicationContext = applicationContext;
        this.applicationId = applicationId;
        this.easyMsMQProperties = easyMsMQProperties;
        this.mqMessageService = mqMessageService;
    }

}
